﻿/// Copyright © 2013 Microsoft Corporation. 
/// This project is released under the Microsoft Public License (MS-PL)
///
namespace DataMorph.Processing
{
    using System;
    using System.Collections.Concurrent;
    using System.Collections.Generic;
    using System.Diagnostics;
    using System.IO;

    using DataMorph.Common.Contracts;
    using DataMorph.Common.DataAccess;
    using DataMorph.Common.Storage;

    using Microsoft.WindowsAzure.Storage.Blob;
    using Microsoft.WindowsAzure.Storage.RetryPolicies;

    public class TagDataWriter : ITagDataWriter
    {
        private readonly IDataAccess dataAccess;

        private readonly string tagId;

        private readonly string deploymentId;

        private readonly string instanceId;

        private readonly CloudBlobContainer container;

        private readonly CloudBlockBlob blob;

        private readonly int batchId;

        private readonly string blobName;

        private readonly ProcessRequest processRequest;

        private readonly string blobFullName;

        private ConcurrentBag<string> lines = new ConcurrentBag<string>();

        public TagDataWriter(ICloudStorageAccount cloudStorageAccount, IDataAccess dataAccess, string tagId, ProcessRequest processRequest, string deploymentId, string instanceId, int batchId)
        {
            this.dataAccess = dataAccess;
            this.tagId = tagId;
            this.deploymentId = deploymentId;
            this.instanceId = instanceId;
            this.batchId = batchId;
            this.processRequest = processRequest;

            var client = cloudStorageAccount.CreateCloudBlobClient();
            this.container = client.GetContainerReference(processRequest.DataSetName);
            this.container.CreateIfNotExists();

            this.blobName = string.Format(
                                    "{0}-{1}-{2}-{3}.bin",
                                    this.tagId,
                                    this.deploymentId,
                                    this.instanceId,
                                    this.batchId);

            this.blobFullName = string.Format("{0}/{1}", processRequest.TimeRequested.ToString("MMddyyyyhhmmss"), this.blobName);

            this.blob = this.container.GetBlockBlobReference(this.blobFullName);
        }

        public void AddTagData(IEnumerable<string> tagData)
        {
            foreach (var line in tagData)
            {
                this.lines.Add(line);
            }
        }

        public WriteTimings Commit()
        {
            lock (this.blob)
            {
                var writeTime = new Stopwatch();
                writeTime.Start();

                this.Write(this.lines);

                writeTime.Stop();

                this.UpdateRecords();

                this.lines = new ConcurrentBag<string>();

                return new WriteTimings
                {
                    WriteTime = writeTime.Elapsed.TotalSeconds,
                    TimeRequested = this.processRequest.TimeRequested.UtcDateTime,
                    StorageAccount = this.processRequest.StorageAccountName,
                    CompletedTime = DateTimeOffset.UtcNow,
                    DataSet = this.processRequest.DataSetName
                };
            }
        }

        private void UpdateRecords()
        {
            this.dataAccess.AddTagBlobRecord(
                this.processRequest.DataSetName,
                this.processRequest.TimeRequested.ToString("MMddyyyyhhmmss"),
                this.tagId,
                this.deploymentId,
                this.instanceId,
                this.batchId,
                this.blob.Uri.AbsoluteUri);
        }

        private void Write(IEnumerable<string> lines)
        {
            using (var memoryStream = new MemoryStream())
            using (var streamWriter = new StreamWriter(memoryStream))
            {
                foreach (var line in lines)
                {
                    streamWriter.WriteLine(line);
                }

                streamWriter.Flush();
                memoryStream.Seek(0, SeekOrigin.Begin);

                try
                {
                    this.blob.UploadFromStream(
                        memoryStream,
                        null,
                        new BlobRequestOptions { RetryPolicy = new ExponentialRetry(TimeSpan.FromSeconds(30), 5) });
                }
                catch (Exception e)
                {
                    Trace.TraceError("Exception while writing resulting blob. {0}", e);
                }
            }
        }
    }
}